#ifndef EVENT_LOOP_H
#define EVENT_LOOP_H


#include <pthread.h>
// #include "ChannelMap.h"
#include "EventDispatcher.h"
#include "Common.h"
#include <assert.h>
#include "EpollDispatcherData.h"
#include <vector>

const int FDEVENTS = 32;
const EventDispatcher epollDispatcher = EventDispatcher();

class EventLoop {

public:
	EventLoop(std:: string threadName = ""):
		threadName_(threadName.empty() ? "main Trhead" : threadName),
		quit_(0),
		eventDispatcher_(&epollDispatcher),
		eventDispatcherData_(new EpollDispatcherData()),
		isHandlePending_(0) {

	    pthread_mutex_init(&mutex_, NULL);
	    pthread_cond_init(&cond_, NULL);

		channelMap_.resize(FDEVENTS, NULL);

		owner_thread_id_ = pthread_self();
	    if (socketpair(AF_UNIX, SOCK_STREAM, 0, socketPair_) < 0) {
    	    printf("socketpair set fialed\n");
	    }

	    pendingHead_ = NULL;
	    pendingTail_ = NULL;

    	Channel *channel = new Channel(socketPair_[1], EVENT_READ, this);
	    channel->setEventReadCallBack(std::bind(&EventLoop::handleWakeup, this));

    	addChannelEvent(socketPair_[1], channel);
	}

	void wakeUp(void) {
	    char one = 'a';
	    ssize_t n = write(socketPair_[0], &one, sizeof(one));
    	if (n != sizeof one) {
        	printf("wakeup event loop thread failed\n");
    	}

	}

	int handleWakeup(void) {
	    char one;
	    size_t n = read(socketPair_[1], &one, sizeof one);
	    if (n != sizeof one) {
	        printf("handleWakeup  failed\n");
	    }

	    printf("wakeup, %s", threadName_.c_str());
	}



	int channelEventActivate(int fd, int revents) {
	    printf("activate channel fd == %d, revents=%d, %s\n", fd, revents, threadName_.c_str());

	    if (fd < 0) {
	    	return 0;
	    }

	    if (fd >= channelMap_.size()) {
			return -1;
	    }

	    Channel *channelPtr = channelMap_[fd];

	    assert(fd == channelPtr->fd);
		std::shared_ptr<void> guard;
		if (channelPtr->tied_) {
			guard = channelPtr->tie_.lock();
			if (guard) {

			    if (revents & EVENT_READ) {
					printf("channel guard read callback start\n");
					if (channelPtr->eventReadCallBack) channelPtr->eventReadCallBack(channelPtr->data);
					printf("channel guard read callback end\n");
			    }

			    if (revents & EVENT_WRITE) {
					printf("channel guard write callback start\n");
					if (channelPtr->eventWriteCallBack) channelPtr->eventWriteCallBack(channelPtr->data);
					printf("channel guard write callback end\n");
			    }
			}
		} else {
		    if (revents & EVENT_READ) {
				printf("channel read callback start\n");
				if (channelPtr->eventReadCallBack) channelPtr->eventReadCallBack(channelPtr->data);
				printf("channel read callback end\n");
		    }

		    if (revents & EVENT_WRITE) {
				printf("channel write callback start\n");
				if (channelPtr->eventWriteCallBack) channelPtr->eventWriteCallBack(channelPtr->data);
				printf("channel write callback end\n");
		    }
		}
	}


    int addChannelEvent(int fd, Channel *channel1) {
    	return doChannelEvent(fd, channel1, 1);
    }


    int removeChannelEvent(int fd, Channel *channel1) {
		return doChannelEvent(fd, channel1, 2);
    }

    int updateChannelEvent(int fd, Channel *channel1) {
    	return doChannelEvent(fd, channel1, 3);
    }

    void channelBufferNoLock(int fd, Channel *channelPtr, int type) {
	    // struct channel_element *channelElement = (channel_element *) malloc(sizeof(struct channel_element));
	    // channelElement->channel = channelPtr;
	    // channelElement->type = type;
	    // channelElement->next = NULL;


	    // if (pendingHead_ == NULL) {
	    // 	pendingHead_ = pendingTail_ = channelElement;
	    // } else {
	    // 	pendingTail_->next = channelElement;
	    // 	pendingTail_ = channelElement;
	    // }
		channelPtr->type = type;
		pendings_.push_back(channelPtr);
    }

	int doChannelEvent(int fd, Channel *channel1, int type) {
		pthread_mutex_lock(&mutex_);

		assert(isHandlePending_ == 0);
		channelBufferNoLock(fd, channel1, type);

		pthread_mutex_unlock(&mutex_);

		if (!(owner_thread_id_ == pthread_self())) {
			wakeUp();
		} else {
			handlePendingChannel();
		}

	}

	int handlePendingAdd(int fd, Channel *channelPtr) {
	    printf("add channel fd == %d, %s\n", fd, threadName_.c_str());

	    if (fd < 0) {
	    	return 0;
	    }

	    if (fd >= channelMap_.size()) {
			int new_size = channelMap_.size();
			while (new_size < fd) {
				 new_size <= 1;
			}
			channelMap_.resize(new_size, NULL);
	    }

	    if (channelMap_[fd] == NULL) {
	    	channelMap_[fd] = channelPtr;
	    	eventDispatcher_->add(eventDispatcherData_, channelPtr);
	    	return 1;
	    }

	    return 0;
	}

	int handlePendingRemove(int fd, Channel *channel1) {

	    assert(fd == channel1->fd);

	    if (fd < 0) {
	    	return 0;
	    }

	    if (fd >= channelMap_.size()) {
	    	return -1;
	    }

		Channel *channel2 = channelMap_[fd];

	    // //update dispatcher(multi-thread)here
	    // struct event_dispatcher *eventDispatcher = eventLoop->eventDispatcher;

	    int retval = 0;
	    if (eventDispatcher_->del(eventDispatcherData_, channel2) == -1) {
	        retval = -1;
	    } else {
	        retval = 1;
	    }

	    channelMap_[fd] = NULL;
	    return retval;

	}

	int handlePendingUpdate(int fd, Channel *channelPtr) {
	    printf("update channel fd == %d, %s\n", fd, threadName_.c_str());

	    assert(fd == channelPtr->fd);

	    if (fd < 0) {
	    	return 0;
	    }

	    if (channelMap_[fd] == NULL) {
	    	return -1;
	    }

	    //update channel
	    eventDispatcher_->update(eventDispatcherData_, channelPtr);

	}

	void handlePendingChannel(void) {
		pthread_mutex_lock(&mutex_);

		isHandlePending_ = 1;

		for (Channel* channel: pendings_) {
			if (channel != NULL) {
				int fd = channel->fd;
				if (channel->type == 1) {
					handlePendingAdd(fd, channel);
				} else if (channel->type == 2) {
					handlePendingRemove(fd, channel);
				} else if (channel->type == 3) {
					handlePendingUpdate(fd, channel);
				}
			}
		}
		pendings_.resize(0);
		// struct channel_element *channelElement = pendingHead_;

		// while (channelElement != NULL) {
		// 	Channel *channel = channelElement->channel;
		// 	int fd = channel->fd;

		// 	if (channelElement->type == 1) {
		// 	    handlePendingAdd(fd, channel);
		// 	} else if (channelElement->type == 2) {
		// 	    handlePendingRemove(fd, channel);
		// 	} else if (channelElement->type == 3) {
		// 	    handlePendingUpdate(fd, channel);
		// 	}
	 //        channelElement = channelElement->next;
		// }

		// pendingHead_ = pendingTail_ = NULL;
		isHandlePending_ = 0;

		pthread_mutex_unlock(&mutex_);

		return;
	}

	/**
	 *
	 * 1.参数验证
	 * 2.调用dispatcher来进行事件分发,分发完回调事件处理函数
	 */
	int run(void) {
	    // assert(eventLoop != NULL);

	    // struct event_dispatcher *dispatcher = eventLoop->eventDispatcher;

	    if (owner_thread_id_ != pthread_self()) {
	        exit(1);
	    }

	    printf("event loop run, %s\n", threadName_.c_str());
	    struct timeval timeval;
	    timeval.tv_sec = 1;

	    while (!quit_) {
			// pendings_.resize(0);
	        //block here to wait I/O event, and get active channels

	        eventDispatcher_->dispatch_loop(this, &timeval);

	        //handle the pending channel
	        handlePendingChannel();
	    }

	    printf("event loop end, %s\n", threadName_.c_str());
	    return 0;
	}

	struct channel_element {
	    int type; //1: add  2: delete
	    Channel *channel;
	    struct channel_element *next;
	};

// private:

	typedef std::vector<Channel*> ChannelMap;
	// typedef std::map<int, Channel*> ChannelElment;

	int quit_;
	const EventDispatcher *eventDispatcher_;
    EpollDispatcherData *eventDispatcherData_;
    ChannelMap channelMap_;
    ChannelMap pendings_;

    int isHandlePending_;
    struct channel_element *pendingHead_;
    struct channel_element *pendingTail_;

    pthread_t owner_thread_id_;
    pthread_mutex_t mutex_;
    pthread_cond_t cond_;
    int socketPair_[2];
    std::string threadName_;
};

#endif
